/**
 * 
 */
package edu.sjtu.seda.eventqueue;

import java.text.MessageFormat;
import java.util.ArrayList;
import java.util.Collection;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.atomic.AtomicBoolean;

import edu.sjtu.seda.event.Event;
import edu.sjtu.seda.event.Event.STATUS;
import edu.sjtu.seda.log.LogService;
import edu.sjtu.seda.stage.StageWrapper;
import edu.sjtu.seda.threadpool.StageRunner;
import edu.sjtu.seda.threadpool.StageRunnerImpl;

/**
 * @author john.zhu
 * 
 */
public class EventQueueImpl implements EventQueue {
	private static LogService log = new LogService(EventQueueImpl.class);

	private AtomicBoolean shutdown = new AtomicBoolean(false);

	private StageWrapper wrapper;

	private int defaultSize = 10;

	private boolean enableBatch;
	private int batchSize = 10;

	public EventQueueImpl(StageWrapper wrapper, int size) {
		this(wrapper, size, false, 0);
	}

	/**
	 * event queue implementation. Receive events, wrap as stagerunner and send
	 * to threadpool finally send to next stage
	 * 
	 * @param wrapper
	 * @param size
	 */
	public EventQueueImpl(StageWrapper wrapper, int size, boolean enableBatch,
			int batchSize) {
		this.wrapper = wrapper;
		this.defaultSize = size;
		this.enableBatch = enableBatch;
		this.batchSize = batchSize;
	}

	public void pushEvent(Event event) {
		if (shutdown.get()) {
			log.error(
					"queue {0} is shutdown and event {1} cannot be pushed to queue",
					wrapper.getStage().getName(), event.getUID());
		} else {
			StageRunner runner = new StageRunnerImpl(event,
					wrapper.getEventHandler(), wrapper);
			log.info("event {0} has been pushed to queue {1}", event.getUID(),
					wrapper.getStage().getName());
			wrapper.getThreadPool().execute(runner);
		}
	}

	@Override
	public void shutdown() {
		shutdown.set(true);
	}

	public boolean isShutdown() {
		return shutdown.get();
	}

}
